#include "Sender.h"
#include <cstring>
#include <unistd.h>
#include <iostream>
#include <arpa/inet.h>
#include "CpuAffinity.h"
#include "RateMonitor.h"

Sender::Sender(const char *target_ip, uint16_t target_port, LockFreeRingBuffer<OrderPacket> &ringBuffer, MemoryPool &memoryPool, int cpu_id)
    : ringBuffer_(ringBuffer), memoryPool_(memoryPool), cpu_id_(cpu_id), running_(false)
{

    sock_ = socket(AF_INET, SOCK_DGRAM, 0);
    if (sock_ < 0)
    {
        perror("[Sender] socket create failed");
        exit(EXIT_FAILURE);
    }

    int optval = 1;
    setsockopt(sock_, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval));
    setsockopt(sock_, SOL_SOCKET, SO_REUSEPORT, &optval, sizeof(optval));

    std::memset(&target_addr_, 0, sizeof(target_addr_));
    target_addr_.sin_family = AF_INET;
    target_addr_.sin_port = htons(target_port);
    target_addr_.sin_addr.s_addr = inet_addr(target_ip);
}

Sender::~Sender()
{
    stop();
    if (sock_ >= 0)
    {
        close(sock_);
    }
}

void Sender::start()
{
    running_ = true;
    thread_ = std::thread(&Sender::run, this);
}

void Sender::stop()
{
    running_ = false;
    if (thread_.joinable())
    {
        thread_.join();
    }
}

void Sender::run()
{
    CpuAffinity::bindThreadToCpu(cpu_id_);

    struct mmsghdr msgs[MAX_BATCH];

    size_t numOrders = MAX_BATCH;
    std::vector<struct iovec> iovs;
    iovs.reserve(numOrders * 8);
    while (running_)
    {
        int batch_size = 0;

        // 拉取RingBuffer中的订单
        size_t iovIndex = 0;
        for (int i = 0; i < MAX_BATCH; ++i)
        {
            OrderPacket *packet = static_cast<OrderPacket *>(memoryPool_.allocate(i));
            if (packet == nullptr)
            {
                std::cerr << "MemoryPool exhausted! Drop packet!" << std::endl;
                return;
            }
            if (!ringBuffer_.pop(*packet))
            {
                break;
            }

            // 打印packet
            printf("send packet: %s %s %s %d %d %llu %u %llu\n", packet->accountId, packet->orderId, packet->symbol, (int)packet->isBuy, packet->orderType, packet->price, packet->quantity, packet->timestamp);

            iovs[iovIndex++] = {packet->accountId, ACCOUNT_ID_SIZE};
            iovs[iovIndex++] = {packet->orderId, ORDER_ID_SIZE};
            iovs[iovIndex++] = {packet->symbol, SYMBOL_SIZE};
            iovs[iovIndex++] = {const_cast<OrderDirection *>(&packet->isBuy), sizeof(packet->isBuy)};
            iovs[iovIndex++] = {const_cast<OrderType *>(&packet->orderType), sizeof(packet->orderType)};
            iovs[iovIndex++] = {const_cast<uint64_t *>(&packet->price), sizeof(packet->price)};
            iovs[iovIndex++] = {const_cast<uint32_t *>(&packet->quantity), sizeof(packet->quantity)};
            iovs[iovIndex++] = {const_cast<uint64_t *>(&packet->timestamp), sizeof(packet->timestamp)};

            msgs[i].msg_hdr.msg_name = &target_addr_;
            msgs[i].msg_hdr.msg_namelen = sizeof(target_addr_);
            msgs[i].msg_hdr.msg_iov = &iovs[i];
            msgs[i].msg_hdr.msg_iovlen = 8;
            msgs[i].msg_hdr.msg_control = nullptr;
            msgs[i].msg_hdr.msg_controllen = 0;
            msgs[i].msg_hdr.msg_flags = 0;
            msgs[i].msg_len = 0; // kernel writeback
            batch_size++;
        }

        if (batch_size > 0)
        {
            int sent = sendmmsg(sock_, msgs, batch_size, 0);
            if (sent < 0)
            {
                perror("[Sender] sendmmsg failed");
                // 可以添加重试逻辑
                int retry_count = 1;
                while (retry_count > 0)
                {
                    sent = sendmmsg(sock_, msgs, batch_size, 0);
                    if (sent >= 0)
                    {
                        break;
                    }
                    perror("[Sender] sendmmsg retry failed");
                    retry_count--;
                }
                if (sent < 0)
                {
                    std::cerr << "[Sender] Failed to send messages after retries." << std::endl;
                }
            }
            else
            {
                printf("[Sender] Sent %d messages.\n", sent);
            }

            // rateMonitor_->incSend();  // sendmmsg 里for循环每发一条调用一次
            // else: 成功发了 sent 条
        }
        // else: 没订单，自旋继续
    }
}
